package com.mgq.juc.threadpool;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author MaGuangQi
 * 实现自定义线程池
 * @date 2021-12-20 16:47
 **/
@Slf4j
public class TestThreadPool {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(1, 1, 1000, TimeUnit.MILLISECONDS,
                (queue, task) -> {
                    //死等
                    //queue.put(task);
                    //放入的时候等待一定时间,失败就返回false
                    //boolean offer = queue.offer(task, 500, TimeUnit.MILLISECONDS);
                    //log.debug("加入队列结果{}",offer);
                    // 放弃任务
                    //log.debug("放弃任务,{}", task);
                    //抛出异常
                   // throw new RuntimeException("抛出异常" + task);
                    //自己执行
                    task.run();
                });
        //创建5个任务,交给线程池执行
        for (int i = 0; i < 4; i++) {
            int j = i;
            threadPool.execute(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.debug("任务" + j);
            });
        }
    }
}

@Slf4j
class ThreadPool {

    /**
     * 任务队列
     */
    private BlockedQueue<Runnable> taskQueues;
    /**
     * 核心线程数
     */
    private int coreSize;
    /**
     * 线程集合
     */
    private HashSet<Work> works = new HashSet<>();

    /**
     * 获取任务时的超时时间
     */
    private long timeout;
    /**
     * 单位
     */
    private TimeUnit unit;

    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int coreSize, int capacity, long timeout, TimeUnit unit, RejectPolicy<Runnable> policy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.taskQueues = new BlockedQueue<>(capacity);
        this.unit = unit;
        this.rejectPolicy = policy;
    }

    public void execute(Runnable task) {
        //线程集合小于核心线程数,直接创建个线程,执行
        //否则将任务赋值给阻塞队列
        synchronized (works) {
            //多个线程执行,works有线程安全问题
            if (works.size() < coreSize) {
                Work work = new Work(task);
                log.debug("新增work,{}", work);
                work.start();
                //将work加入线程集合
                works.add(work);
            } else {
                taskQueues.tryPut(rejectPolicy, task);
                //不能直接执行rejectPolicy.reject(taskQueues, task);因为队列可能没有满,直接执行策略,比如放弃时,
                // 会导致未加入队列,就放弃了
                //rejectPolicy.reject(taskQueues, task);
                //taskQueues.put(task);
                //如果队列满了,此时如果直接调用put,会让主线程死等,为了解决这个问题,引入策略
                //1. 主线程可以选择死等
                //2. 可以选择超时等待
                //3. 可以放弃任务
                //4. 可以抛出异常
                //5. 可以选择自己把这个任务执行了
                //因此,我们使用策略模式,将选择提供给调用方.由调用方来选择
            }
        }
    }

    class Work extends Thread {
        /**
         * 任务对象
         */
        private Runnable task;

        public Work(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            //执行任务
            //当task不为空时,执行任务
            //当任务队列不为空时,执行队列中的任务,take()会阻塞等待,如果需要超时等待,就可以用poll
            //while (task != null ||(task= taskQueues.take()) != null) {
            while (task != null || (task = taskQueues.poll(timeout, unit)) != null) {
                //执行任务.catch住异常,因为任务有可能抛出异常
                try {
                    log.debug("任务正在被执行,{}", task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    task = null;
                }
            }
            //退出循环,从集合中移除this
            synchronized (works) {
                log.debug("work被移除,{}", this);
                works.remove(this);
            }
        }
    }
}

@FunctionalInterface
interface RejectPolicy<T> {
    public void reject(BlockedQueue<T> queue, T task);
}

@Slf4j
class BlockedQueue<T> {
    //队列容量
    private int capacity;
    //双向链表
    private Deque<T> deque;

    //锁
    private ReentrantLock lock = new ReentrantLock();
    //满的条件变量
    private Condition fullWaitSet = lock.newCondition();
    //队列空等待
    private Condition emptyWaitSet = lock.newCondition();

    public BlockedQueue(int capacity) {
        this.capacity = capacity;
        deque = new ArrayDeque<>(capacity);
    }

    public T poll(long timeout, TimeUnit unit) {
        lock.lock();
        try {

            long nanos = unit.toNanos(timeout);
            while (deque.isEmpty()) {
                try {
                    if (nanos <= 0L)
                        return null;
                    //这个方法的返回值是等待的时间-已经花费的时间
                    //因此,我们只需要将结果重新赋值给nanos
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = deque.removeFirst();
            fullWaitSet.signalAll();
            return t;
        } finally {
            lock.unlock();
        }
    }

    /**
     * take 不带超时时间
     *
     * @return
     */
    public T take() {
        lock.lock();
        try {
            //使用while, 不要用if
            while (deque.isEmpty()) {
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = deque.removeFirst();
            fullWaitSet.signalAll();
            return t;

        } finally {
            lock.unlock();
        }
    }

    /**
     * 放入元素
     *
     * @param element
     */
    public void put(T element) {
        lock.lock();
        try {
            while (deque.size() == capacity) {
                try {
                    log.debug("队列已满,等待加入任务队列,{}", element);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("将任务加入队列,{}", element);
            deque.addLast(element);
            emptyWaitSet.signalAll();

        } finally {
            lock.unlock();
        }
    }

    /**
     * 带超时时间放入元素
     *
     * @param element
     */
    public boolean offer(T element, long timeout, TimeUnit unit) {
        long nanos = unit.toNanos(timeout);
        lock.lock();
        try {
            while (deque.size() == capacity) {
                try {
                    if (nanos <= 0) {
                        return false;
                    }
                    log.debug("等待加入任务队列.............,{}", element);
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            deque.addLast(element);
            emptyWaitSet.signalAll();
            return true;
        } finally {
            lock.unlock();
        }
    }

    public int getSize() {
        lock.lock();
        try {
            return capacity;
        } finally {
            lock.unlock();
        }

    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            //队列没满
            if (deque.size() != capacity) {
                log.debug("将任务加入队列,{}", task);
                deque.addLast(task);
                emptyWaitSet.signalAll();
            } else {
                //满了,执行策略
                rejectPolicy.reject(this, task);
            }
        } finally {
            lock.unlock();
        }


    }
}
